#include "filters/filter_worker.h"
#include "utils/logger.h"

#include <chrono>

FilterWorker::FilterWorker(size_t worker_id, std::shared_ptr<IFilterEngine> filterEngine, PacketDispatcher::PacketQueue& inputQueue)
    : m_workerId(worker_id), m_filterEngine(std::move(filterEngine)), m_inputQueue(inputQueue){
    LOG_DEBUG("FilterWorker " << worker_id << " created");
}

FilterWorker::~FilterWorker() {
    stop();
    LOG_DEBUG("FilterWorker " << m_workerId << " destroyed");
}

bool FilterWorker::start() {
    if (m_running.load(std::memory_order_acquire)) {
        LOG_WARNING("FilterWorker " << m_workerId << " already running");
        return true;
    }

    m_shouldStop.store(false, std::memory_order_release);
    m_workerThread = std::thread(&FilterWorker::workerThread, this);
    m_running.store(true, std::memory_order_release);

    LOG_INFO("FilterWorker " << m_workerId << " started");
    return true;
}

void FilterWorker::stop() {
    if (!m_running.load(std::memory_order_acquire)) return;

    m_shouldStop.store(true, std::memory_order_release);
    if (m_workerThread.joinable()) {
        m_workerThread.join();
    }
    m_running.store(false, std::memory_order_release);

    LOG_INFO("FilterWorker " << m_workerId << " stopped");
}

void FilterWorker::workerThread() {
    LOG_DEBUG("FilterWorker " << m_workerId << " thread started");

    std::vector<Packet> batch;
    batch.reserve(m_batchSize);

    uint64_t emptyCycles = 0;

    while (!m_shouldStop.load(std::memory_order_acquire)) {
        /// Try batch process
        size_t count = m_inputQueue.pop_bulk(batch.data(), m_batchSize);

        if (count > 0) {
            emptyCycles = 0;
            auto startTime = std::chrono::high_resolution_clock::now();

            /// Process a batch of packets
            for (size_t i = 0; i < count; ++i) {
                FilterResult result = m_filterEngine->processPacket(batch[i]);

                /// Update statistics
                {
                    std::lock_guard<std::mutex> lock(m_statsMutex);
                    m_stats.processed_packets++;
                    if (result.packet_allowed) {
                        m_stats.accepted_packets++;
                    } else {
                        m_stats.dropped_packets++;
                    }
                }

                /// If the packet is accepted and there is a output callback, call callback function
                if (result.packet_allowed && m_outputCallback) {
                    m_outputCallback(std::move(batch[i]), result);
                }
                /// If the packet is dropped, it will be destroyed when leaving batch[i]
            }

            auto endTime = std::chrono::high_resolution_clock::now();
            auto processingTime = std::chrono::duration_cast<std::chrono::nanoseconds>(endTime - startTime).count();
            m_totalProcessingTime += processingTime;

            {
                std::lock_guard lock(m_statsMutex);
                m_stats.batch_processed++;
            }
        } else {
            emptyCycles++;
            handleEmptyQueue();

            {
                std::lock_guard<std::mutex> lock(m_statsMutex);
                m_stats.queue_empty_cycles++;
            }
        }

        /// Periodically update statistics on queue size
        if (emptyCycles % 10 == 0) {
            std::lock_guard lock(m_statsMutex);
            m_stats.current_queue_size = m_inputQueue.size();
        }
    }

    LOG_DEBUG("FilterWorker " << m_workerId << " thread exiting");
}

void FilterWorker::handleEmptyQueue() {
    constexpr uint64_t YIELD_THRESHOLD = 100;
    constexpr uint64_t SLEEP_THRESHOLD = 1000;

    static thread_local uint64_t consecutive_empty_cycle = 0;
    consecutive_empty_cycle++;
    if (consecutive_empty_cycle < YIELD_THRESHOLD) {
        /// Briefly spin and the yield the CPU
        std::this_thread::yield();
    } else if (consecutive_empty_cycle < SLEEP_THRESHOLD) {
        /// Moderate spinning followed by brief sleep
        std::this_thread::sleep_for(std::chrono::microseconds(10));
    } else {
        /// Extended spinning followed by longer sleep
        std::this_thread::sleep_for(std::chrono::microseconds(100));
        consecutive_empty_cycle = 0;    /// Reset the count
    }
}

FilterWorker::WorkerStats FilterWorker::getStats() const {
    std::lock_guard<std::mutex> lock(m_statsMutex);
    WorkerStats stats = m_stats;
    stats.worker_id = m_workerId;

    /// Calculate average process time
    if (stats.processed_packets > 0) {
        stats.average_processing_time_us = static_cast<double>(m_totalProcessingTime.load()) / stats.processed_packets / 1000.0;
    }

    stats.current_queue_size = m_inputQueue.size();

    return stats;
}